package com.xormedia.httpservice.util;

/**
 * a class to throttle the data transfers. This class is thread safe. It can be
 * shared by multiple threads. The parameter bandwidthPerSec specifies the total
 * bandwidth shared by threads.
 */
public class DataTransferThrottler {
  // 周期
  private final long period; // period over which bw is imposed
  private final long periodExtension; // Max period over which bw accumulates.
  // 一个周期内可以发送/接收的byte总数
  private long bytesPerPeriod; // total number of bytes can be sent in each
                               // period
  // 当前周期开始时间，单位毫秒
  private long curPeriodStart; // current period starting time
  // 当前周期内还可以发送/接收的byte数
  private long curReserve; // remaining bytes can be sent in the period
  // 当前周期内已经使用的byte数
  private long bytesAlreadyUsed;

  /**
   * Constructor
   * 
   * @param bandwidthPerSec
   *          bandwidth allowed in bytes per second.
   */
  public DataTransferThrottler(long bandwidthPerSec) {
    this(500, bandwidthPerSec); // by default throttling period is 500ms
  }

  /**
   * Constructor
   * 
   * @param period
   *          in milliseconds. Bandwidth is enforced over this period.
   * @param bandwidthPerSec
   *          bandwidth allowed in bytes per second.
   */
  public DataTransferThrottler(long period, long bandwidthPerSec) {
    this.curPeriodStart = monotonicNow();
    this.period = period;
    // 将带宽按照周期做比例转化
    this.curReserve = this.bytesPerPeriod = bandwidthPerSec * period / 1000;
    this.periodExtension = period * 3;
  }

  /**
   * Current time from some arbitrary time base in the past, counting in
   * milliseconds, and not affected by settimeofday or similar system clock
   * changes. This is appropriate to use when computing how much longer to wait
   * for an interval to expire.
   * 
   * @return a monotonic clock that counts in milliseconds.
   */
  public static long monotonicNow() {
    final long NANOSECONDS_PER_MILLISECOND = 1000000;

    return System.nanoTime() / NANOSECONDS_PER_MILLISECOND;
  }

  /**
   * @return current throttle bandwidth in bytes per second.
   */
  public synchronized long getBandwidth() {
    return bytesPerPeriod * 1000 / period;
  }

  /**
   * Sets throttle bandwidth. This takes affect latest by the end of current
   * period.
   */
  public synchronized void setBandwidth(long bytesPerSecond) {
    if (bytesPerSecond <= 0) {
      throw new IllegalArgumentException("" + bytesPerSecond);
    }
    bytesPerPeriod = bytesPerSecond * period / 1000;
  }

  /**
   * Given the numOfBytes sent/received since last time throttle was called,
   * make the current thread sleep if I/O rate is too fast compared to the given
   * bandwidth.
   *
   * @param numOfBytes
   *          number of bytes sent/received since last time throttle was called
   */
  public synchronized void throttle(long numOfBytes) {
    throttle(numOfBytes, null);
  }

  /**
   * Given the numOfBytes sent/received since last time throttle was called,
   * make the current thread sleep if I/O rate is too fast compared to the given
   * bandwidth. Allows for optional external cancelation.
   *
   * @param numOfBytes
   *          number of bytes sent/received since last time throttle was called
   * @param canceler
   *          optional canceler to check for abort of throttle
   */
  public synchronized void throttle(long numOfBytes, Canceler canceler) {
    if (numOfBytes <= 0) {
      return;
    }

    // 当前的可传输的字节数减去当前发送/接收字节数
    curReserve -= numOfBytes;
    // 当前字节使用量
    bytesAlreadyUsed += numOfBytes;

    // 如果curReserve<=0,说明当前周期内可使用字节数已经用完
    while (curReserve <= 0) {
      // 如果设置了canceler对象,则不会进行限流操作
      if (canceler != null && canceler.isCancelled()) {
        return;
      }
      long now = monotonicNow();
      long curPeriodEnd = curPeriodStart + period;
      // 如果当前时间还在本周期时间内的话,则必须等待此周期的结束,重新获取新的可传输字节量
      if (now < curPeriodEnd) {
        // Wait for next period so that curReserve can be increased.
        try {
          wait(curPeriodEnd - now);
        } catch (InterruptedException e) {
          // Abort throttle and reset interrupted status to make sure other
          // interrupt handling higher in the call stack executes.
          Thread.currentThread().interrupt();
          break;
        }
      } else if (now < (curPeriodStart + periodExtension)) {
        // 如果当前时间已经超过此周期的时间且不大于最大周期间隔,则增加可接受字节数,
        // 并更新周期起始时间为前一周期的末尾时间
        curPeriodStart = curPeriodEnd;
        curReserve += bytesPerPeriod;
      } else {
        // 如果当前时间已经超过此周期的时间且不大于最大周期间隔,则增加可接受字节数,
        // 并更新周期起始时间为前一周期的末尾时间
        // discard the prev period. Throttler might not have
        // been used for a long time.
        curPeriodStart = now;
        curReserve = bytesPerPeriod - bytesAlreadyUsed;
      }
    }
    // 传输结束,当前字节使用量进行移除
    bytesAlreadyUsed -= numOfBytes;
  }
}
